package org.dbstar.scheduler.job;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.dbstar.scheduler.beans.CompoundJob;
import org.dbstar.scheduler.beans.CompoundJobConf;
import org.dbstar.scheduler.beans.CompoundJobResult;
import org.dbstar.scheduler.beans.Job;
import org.dbstar.scheduler.beans.JobConf;
import org.dbstar.scheduler.beans.JobDefine;
import org.dbstar.scheduler.control.JobController;
import org.dbstar.scheduler.control.exception.DaoException;
import org.dbstar.scheduler.identifier.Identifier;
import org.dbstar.scheduler.identifier.IdentifierFactory;
import org.dbstar.scheduler.identifier.IdentifierType;
import org.dbstar.scheduler.utils.JobUtils;


/**
 * 构建CompoundJob
 * 
 * @author 代波
 * @since 1.0.0
 * @version 1.0.0
 * @date 2012-5-7
 */
public class CompoundJobFactory extends AbstractJobFactory<CompoundJob> {
	private final JobController jobController;
	private final IdentifierFactory identifierFactory;

	public CompoundJobFactory(JobController jobController, IdentifierFactory identifierFactory) {
		super();
		this.jobController = jobController;
		this.identifierFactory = identifierFactory;
	}

	@Override
	public CompoundJob createJob(SchedulerFactory schedulerFactory, JobConf conf, String submitter,
			Map<String, String> parameters, long submitTime) throws IllegalFactoryException {
		if (conf == null) {
			throw new IllegalFactoryException("Job Conf is null");
		}
		if (StringUtils.isBlank(submitter)) {
			throw new IllegalFactoryException("Job Submitter is null");
		}
		CompoundJob job = new CompoundJob();
		job.setJobConfIdentifier(conf.getIdentifier());
		job.setSubmitTime(submitTime);
		job.setSubmitter(submitter);
		job.setParameters(JobUtils.getJobParameters(conf, parameters, true));
		try {
			build(job, (CompoundJobConf) conf, submitter, schedulerFactory);
		} catch (DaoException e) {
			throw new IllegalFactoryException(e);
		}
		return job;
	}

	private void build(CompoundJob job, CompoundJobConf conf, String submitter, SchedulerFactory schedulerFactory)
			throws IllegalFactoryException, DaoException {
		List<JobDefine> list = new ArrayList<JobDefine>();
		CompoundJobResult result = (CompoundJobResult) job.getResult();
		for (JobDefine jd : conf.getJobs()) {
			boolean skip = !JobUtils.matchCondition(jd.getCondition(), job.getParameters());
			if (skip) {
				Identifier identifier = identifierFactory.createIdentifier(IdentifierType.skip);
				list.add(new JobDefine(identifier, JobUtils.formatSubJobParameters(jd.getParameters(),
						job.getParameters()), jd.getDependencies(), jd.getCondition(), skip));
				result.addSubJobResult(identifier, JobStatus.init, false, jd.getDependencies());
			} else {
				Job subJob = schedulerFactory.createJob(jd.getIdentifier(),
						JobUtils.formatSubJobParameters(jd.getParameters(), job.getParameters()), submitter);
				list.add(new JobDefine(jobController.addJob(subJob), subJob.getParameters(), jd.getDependencies(), jd
						.getCondition(), skip));
				result.addSubJobResult(subJob, jd.getDependencies());
			}
		}
		job.setSubJobs(list);
	}
}
